package cyou.breathe.blog.mq.consumer;

import com.alibaba.fastjson2.JSON;
import cyou.breathe.blog.mq.param.MessageBody;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;

import static cyou.breathe.blog.mq.producer.StreamProducer.*;

/**
 * @author: breathe
 * @createTime: 2025-08-26
 */
public class AbstractStreamConsumer {

    private static Logger LOG = LoggerFactory.getLogger(AbstractStreamConsumer.class);
    public static <T> T getMessage(Message<MessageBody> msg, Class<T> clazz) {
        String messageId = msg.getHeaders().get(ROCKET_MQ_MESSAGE_ID, String.class);
        String tag = msg.getHeaders().get(ROCKET_TAGS, String.class);
        String topic = msg.getHeaders().get(ROCKET_MQ_TOPIC, String.class);
        T result = JSON.parseObject(msg.getPayload().getBody(), clazz);
        LOG.info("Received Message topic:{} messageId:{},object:{}，tag:{}", topic, messageId, JSON.toJSONString(result), tag);
        return result;
    }
}

